package com.cumt.tool.kafka;

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Properties;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.locks.ReentrantLock;

/**
 * Created by caicf on 2017/1/12.
 */
public class KafkaProducerUtil {

    private static final Logger log = LoggerFactory.getLogger(KafkaProducerUtil.class);

    private static ReentrantLock lock = new ReentrantLock();

    private static KafkaProducer<String, String> producer;

    private static KafkaProducerUtil instance;

    private KafkaProducerUtil() {

    }

    public static KafkaProducerUtil getInstance() {
        if (instance == null) {
            lock.lock();
            if (instance == null) {
                instance = new KafkaProducerUtil();
            }
            lock.unlock();
        }
        return instance;
    }

    private KafkaProducer<String, String> getProducer() {
        if (producer == null) {
            lock.lock();

            Properties props = new Properties();
            props.put("bootstrap.servers", "192.168.56.106:9092");
            props.put("acks", "1");
            props.put("retries", 0);
            props.put("batch.size", 16384);
            props.put("key.serializer", StringSerializer.class.getName());
            props.put("value.serializer", StringSerializer.class.getName());
            producer = new KafkaProducer<>(props);

            lock.unlock();
        }
        return producer;
    }

    public void closeProducer() {
        producer.close();
    }

    private static ExecutorService exec = Executors.newCachedThreadPool();

    public boolean sendMessage(String topic, Integer partition, String key, String value) {
        boolean flag = true;
        KafkaProducer<String, String> producer = getProducer();
        ProducerRecord<String, String> record = new ProducerRecord<>(topic, partition, key, value);
        try {
            producer.send(record, (metadata, e) -> {
                if (e != null)
                    e.printStackTrace();
                log.info("message send to partition: {} , offset: {}", metadata.partition(), metadata.offset());
            });
        } catch (Exception e) {
            flag = false;
            log.info("exception {}", e.getMessage());
        }
        return flag;
    }

    public boolean sendMessage(String topic, String key, String value) {
        return sendMessage(topic, null, key, value);
    }

    public boolean sendMessage(String topic, String value) {
        return sendMessage(topic, null, null, value);
    }

}
